
#include "RealTimeDealList.h"
#include "RealTimeDealCache.h"
#include "../BondMarketStream.h"
#include "../JumpListNew.h"
#include <core/time/time.h>

#if WIN32
#include <windows.h>
#else
#endif
#include "RealTimeDealStat.h"

#ifndef BASSERTS
#define BASSERTS // TODO 
#endif // !BASSERTS

#ifndef max
#define max(a,b)            (((a) > (b)) ? (a) : (b))
#endif

typedef MarketStreamInfo		Record;
namespace detail{
	class RecordMap : public IRecordMap{
		typedef std::map<DealIndex, Record> IndexRecordMap;
		IndexRecordMap		m_values;
	public:
		virtual ~RecordMap(){}
		virtual void Enumerate(Callback* cb)const
		{
			IndexRecordMap::const_iterator it = m_values.cbegin();
			for (; it != m_values.cend(); ++it)
			{
				cb->OnEach(it->first, &it->second);
			}
		}
		virtual const void* FindRecord(const DealIndex& id)const
		{
			IndexRecordMap::const_iterator it = m_values.find(id);
			if (it == m_values.end())
				return nullptr;
			return &it->second;
		}
		virtual void Add(const DealIndex& id, const void* record)
		{
			const Record* precord = (const Record*)record;
			IndexRecordMap::iterator it = m_values.find(id);
			if (it == m_values.end())
				m_values.insert(IndexRecordMap::value_type(id, *precord));
			else
				it->second = *precord;
		}
		virtual void Remove(const DealIndex& id)
		{
			// m_values中不能删除,UI中的RealTimeDealModel还保留这对应id
			IndexRecordMap::const_iterator it = m_values.find(id);
			if (it != m_values.end())
			{
				m_values.erase(it);
			}
		}
	};
	class RTDealTimeComparator{
		IRecordMap*			m_values;
	public:
		RTDealTimeComparator(IRecordMap* values) :m_values(values){}
		void SetValues(IRecordMap* values){ m_values = values; }
		RTDealTimeComparator(const RTDealTimeComparator& cmp) :m_values(cmp.m_values){}
		RTDealTimeComparator& operator=(const RTDealTimeComparator& cmp){
			m_values = cmp.m_values;
			return *this;
		}
		int operator()(const DealIndex& b1, const DealIndex& b2)const{
			// max(obj.m_create_time, obj.m_modify_time) 较大的排在前面
			const void* v1 = m_values->FindRecord(b1);
			const void* v2 = m_values->FindRecord(b2);
			if (v1 && v2)
			{
				const Record& r1 = *((const Record*)v1);
				const Record& r2 = *((const Record*)v2);
				time_t t1 = max(r1.m_create_time, r1.m_modify_time);
				time_t t2 = max(r2.m_create_time, r2.m_modify_time);
				// 参见MarketStreamInfo的operator<
				return t1 == t2 ? (r1.m_indexID > r2.m_indexID ? -1 : 1) : (t1 > t2 ? -1 : 1);
			}
			if (v1 == nullptr)
			{
				//BLOGERR("[STREAM][INVALID]can not find DealIndex:%s", b1.id.c_str());
				return 1;
			}
			if (v2 == nullptr)
			{
				//BLOGERR("[STREAM][INVALID]can not find DealIndex:%s", b2.id.c_str());
				return -1;
			}
			return 0;
		}
		int GetExpandLength(int count, int length){ return length * 2; }
	};

	typedef JumpList<DealIndex, RTDealTimeComparator> BondDealJumpListBase;
	class CBondDealJumpList : public BondDealJumpListBase, public IBatchedKeys{
		struct Operation{
			int			indexid : 30;	// MarketStreamInfo.m_indexID
			int			optype : 2;		// 操作类型:Null/Add/Erase/Update
			Operation*	next;			// 单链表
		};
        typedef std::map<int, Operation*> OperationMap;		// key:batchid;
		int			m_batch_id;			// 当前的批此id
	public:
		CBondDealJumpList(RTDealTimeComparator* comparator = 0, int stepCount = 40, int hashInitSize = 2560) :
			BondDealJumpListBase(comparator, stepCount, hashInitSize),
			m_batch_id(-1)
		{}
		virtual int	GetBatchId()const{ return m_batch_id; }
		virtual void SetBatchId(int id){
			if (id >= m_batch_id)
				m_batch_id = id;
		}
        virtual void GetKeys(std::vector<DealIndex>& allkeys)const
		{
			allkeys.reserve(GetCount());
			return GetAllKeys(allkeys);
		}
	};

	class BondKeyDealIndexMap{
        typedef std::string		BondKeyType;
        class DealIndexSet : protected std::set<DealIndex>{
			DealIndex				m_head;			// >1的时候有效
			bool					m_dirty;		// GetHead获取的时候需要更新
		public:
			DealIndexSet() :m_dirty(true){}
			void Add(const DealIndex& id, IRecordMap* values)
			{
				insert(id);
				m_dirty = true;
			}
			void Remove(const DealIndex& id, IRecordMap* values)
			{
				iterator it = find(id);
				if (it != end())
					erase(it);
				m_dirty = true;
			}
			bool GetHead(DealIndex& id, IRecordMap* values)
			{
				if (size() > 1 && m_dirty)
				{
					UpdateHead(values);
				}
				m_dirty = false;
				if (empty())return false;
				if (size() == 1){ id = *begin(); return true; }
				id = m_head;
				BASSERTS(find(id) != end(), "");
				return true;
			}
		protected:
			void UpdateHead(IRecordMap* values)
			{
				if (size() <= 1 || values == nullptr)return;
				int index = 0;
				const void* last = nullptr;
				for (iterator it = begin(); it != end(); ++it, ++index)
				{
					const DealIndex& cur = *it;
					const void* cur_record = values->FindRecord(cur);
					if (cur_record)
					{
						if (!last)//第一个
						{
							m_head = cur;
							last = cur_record;
						}
						else// 挑选排列在最前面的,即时间最大,m_indexID最大
						{
							const Record& r1 = *((const Record*)last);
							const Record& r2 = *((const Record*)cur_record);
							time_t t1 = max(r1.m_create_time, r1.m_modify_time);
							time_t t2 = max(r2.m_create_time, r2.m_modify_time);
							if (t2 > t1 || (t2 == t1 && r2.m_indexID > r1.m_indexID))
							{
								m_head = cur;
								last = cur_record;
							}
						}
					}
				}
				BASSERTS(find(m_head) != end(), "");
			}
		};
        typedef std::map<BondKeyType, DealIndexSet*> BondIndexMap;
		BondIndexMap		m_index_map;
	public:
		BondKeyDealIndexMap(){  }
		~BondKeyDealIndexMap(){
			for (BondIndexMap::iterator it = m_index_map.begin(); it != m_index_map.end(); ++it)
			{
				DealIndexSet* diset = it->second;
				if (diset)delete diset;
			}
			m_index_map.clear();
		}
		void Add(const char * szBondKey, const  char * szListedMarket, const char* szCompanyId, const DealIndex& id, IRecordMap* values)
		{
			char keybuf[128] = { 0 };
			FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
			BondIndexMap::iterator it = m_index_map.find(keybuf);
			DealIndexSet* diset = nullptr;
			if (it == m_index_map.end())
			{
				diset = new DealIndexSet();
				m_index_map.insert(BondIndexMap::value_type(keybuf, diset));
			}
			else
				diset = it->second;
			diset->Add(id, values);
		}
		void Remove(const char * szBondKey, const  char * szListedMarket, const char* szCompanyId, const DealIndex& id, IRecordMap* values)
		{
			char keybuf[128] = { 0 };
			FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
			BondIndexMap::iterator it = m_index_map.find(keybuf);
			if (it != m_index_map.end())
			{
				DealIndexSet* diset = it->second;
				diset->Remove(id, values);
			}
		}
		bool Find(const char * szBondKey, const  char * szListedMarket, const char* szCompanyId, DealIndex& id, IRecordMap* values)const
		{
			char keybuf[128] = { 0 };
			FMTBUF(keybuf, "%s.%s-%s", szBondKey, szListedMarket, szCompanyId);
			BondIndexMap::const_iterator it = m_index_map.find(keybuf);
			if (it == m_index_map.end())
				return false;

			// 找出diset的几个之中,time最大的那个
			const DealIndexSet* diset = it->second;
			bool suc = const_cast<DealIndexSet*>(diset)->GetHead(id, values);
			return suc;
		}
	};
	RealTimeDealList::RealTimeDealList(DayType day, int64 state) :m_tag(OBJECT_TAG), m_day(day), m_bonds(nullptr), m_check(false),
		m_state(state)
	{
		m_values = new detail::RecordMap();
		m_keys = new detail::CBondDealJumpList();
		m_stats = new detail::RealTimeDealStat();
		m_cache = new detail::RealTimeDealCache(m_state);
		m_bonds = new BondKeyDealIndexMap();


		// 必须设置比较器,否则m_keys无法插入数据,插入会崩溃
		m_keys->SetComparator(new detail::RTDealTimeComparator(m_values));
	}
	RealTimeDealList::~RealTimeDealList()
	{
		if (m_values)	delete m_values;
		if (m_keys)		delete m_keys;
		if (m_stats)	delete m_stats;
		if (m_cache)	delete m_cache;
		if (m_bonds)	delete m_bonds;
		m_values = nullptr;
		m_keys = nullptr;
		m_stats = nullptr;
		m_cache = nullptr;
		m_bonds = nullptr;
	}

	int RealTimeDealList::GetRecords(IRealTimeDealModel* keys, int& pos, int count, void* records)
	{
		// UI线程:此接口用于界面表格的一屏幕记录
		// 当日成交中还有一个需要访问所有记录,判断是否是只有一个broker;
		// 这种需求需要访问所有记录,重复计算的是哪一部分?
		// 不管怎么样,始终获取count个记录,
		// 如果范围段[pos,pos+count)中有DealIndex不存在,那么往前往后继续遍历

		// 1.首先,将keys与m_keys同步,如果严重滞后,则重建并过滤
		bool synchronized = false;
		int nOldCount = keys->GetCount();
		class SyncCallback : public detail::RealTimeDealCache::Callback{
			IRealTimeDealModel*	m_keys;
		public:
			SyncCallback(IRealTimeDealModel* keys) :m_keys(keys){}
			virtual bool OnRecord(int id, const Record* record){
				m_keys->AddRecord(id, record);
				return true;
			}
		};
		class FlashCallback : public detail::RealTimeDealCache::Callback{
			IRealTimeDealModel*	m_keys;
		public:
			FlashCallback(IRealTimeDealModel* keys) :m_keys(keys){}
			virtual bool OnRecord(int id, const Record* record){
				m_keys->AddFlash(id, record);
				return true;
			}
		};

		int batchid = keys->GetBatchId();
		if (!IsObjectValid())
			return 0;

		bool equal_state = keys->GetState() == m_cache->GetState();
		if (equal_state && batchid >= 0 && batchid == m_cache->GetEndId())
		{
			// 已经是同步的,无需再次同步
		}
		else if (equal_state && batchid >= 0 && batchid >= m_cache->GetStartId() &&
			batchid < m_cache->GetEndId())
		{
			// 快速同步已经做改进,将没有更新的合并,并分为两类:删除和添加,并维持原有顺序,
			// 然后先删除所有该删除的,再添加所有该添加的
			SyncCallback callback(keys);
			keys->StartCompare(m_values);
			m_cache->GetSince(batchid, callback);
		}
		else//全量同步
		{
			keys->SetState(m_cache->GetState());
			keys->ResetKeys();
			synchronized = true;
			keys->SyncKeys(m_keys, m_values);//比较耗时,
			if (equal_state && batchid >= 0 && batchid <= m_cache->GetEndId())//batchid为-1表示全部重拉,无需重建闪烁,全部清空闪烁
			{
				FlashCallback flashcb(keys);
				m_cache->GetSince(batchid, flashcb);
			}
			else
				keys->ClearFlash();
		}


		//2.更新获取行位置,快速
		int nNewCount = keys->GetCount();
		if (nNewCount != nOldCount || synchronized)
		{
			// pos范围需要重新调整,直至满足要求pos+count<nNewCount
			// nNewCount 是最新数据的总个数,count是界面表格能显示的个数
			// 返回值是实际获取到的记录个数,肯定<=count
		}
		int nret = 0;
		DealIndex indexes[512];//一屏幕最大显示多少行,512是绝对足够的

		DealIndex* pIndex = nullptr;
		bool bNeedDel = false;
		if (count < 512){
			//3.获取一屏幕的记录,通常只有30-50个左右,很快
			pIndex = indexes;
		}
		else {
			bNeedDel = true;

			pIndex = new DealIndex[count];
		}

		int nadded = 0;		// 通常最终返回,nadded==nret
		keys->StartCompare(m_values);
		nret = keys->GetKeys(pos, count, indexes);

		Record* precords = (Record*)records;
		for (int i = 0; i < nret; i++)
		{
			const Record* record = (const Record*)m_values->FindRecord(indexes[i]);
			BASSERTS(record != nullptr, "m_values can not find key:%d", indexes[i]);
			if (record)
				precords[nadded++] = *record;
		}

		//4.检查是否有全部记录的特殊标志:多个broker,用于控制broker列的显示和隐藏
		// 比较耗时,需要访问所有的values(过滤后的),但是只有在OnFilter中,才会将multidirty设置为true
		// synchronized 为false时,已经通过GetSince,AddRecord来更新到m_multitest了;
		if (synchronized && keys->IsMultiEnable())
		{
			keys->CheckMulti(m_values);
		}
		if (bNeedDel) delete[] pIndex;
		return nadded;
	}

	IRecordMap* RealTimeDealList::GetValues(){ return m_values; }
    IRecordMap* RealTimeDealList::GetKeys(std::vector<DealIndex>& keys, bool all)const
	{
		// 获取当前的所有成交id的数组
		// all:false 表示获取排除作废成交的成交id的数组
		keys.clear();
		if (all)
		{
			m_keys->GetKeys(keys);
		}
		else
		{
            std::vector<DealIndex> allkeys;
			m_keys->GetKeys(allkeys);
			keys.reserve(allkeys.size());
			for (int i = 0; i < (int)allkeys.size(); i++)
			{
				const MarketStreamInfo* msi = (const MarketStreamInfo*)m_values->FindRecord(allkeys[i]);
				if (msi && !msi->IsDealStatusCanceled())
					keys.push_back(allkeys[i]);
			}
		}
		return m_values;
	}
	int RealTimeDealList::GetVersion(int companyID) const
	{
		std::map<int, int>::const_iterator it = m_mapversion.find(companyID);
		if (it == m_mapversion.end())return 0;
		return it->second;
	}
	void RealTimeDealList::InsertDirectly(const MarketStreamInfo& info, uint32_t unDay, int batchid)
	{
		BASSERTS(unDay == m_day, "[DEAL]m_day=%d  unDay:%d", m_day, unDay);
		int nDealStatus = atoi(info.m_body.m_dealStatus);
		/*BLOGINFO("[STREAM][RECV]m_indexID:%d	m_id:%s  count:%d code:%s key:%s market:%s status:%s mtime:%d", info.m_indexID, info.m_body.m_id, m_keys?m_keys->GetCount():0,
			info.m_body.m_bondCode,info.m_body.m_bondkey,info.m_body.m_listedmarket,info.m_body.m_dealStatus,info.m_modify_time
			);
			*/
		DealIndex id(info.m_body.m_id);
		/*
			此处更新涉及如下5个模块:
			m_keys	 :快速有序队列
			m_values :成交记录表
			m_stats  :成交统计表
			m_bonds  :债券到成交索引的表
			m_cache  :推送命令缓存

			m_keys(快速有序队列):
			是一个存储MarketStreamType2Body.m_id的有序队列,顺序按照RTDealTimeComparator来比较
			具体的就是MarketStreamInfo的时间m_modify_time|m_create_time较大的排在前面,
			时间相同,再比较m_indexID,较大的排在前面
			快速查找:哈希定位,复杂度O(1)
			删除移除:log(N)
			插入操作:Insert:先删除,后找到合适的排序位置,插入,要求键值对应的记录必须到位
			依赖m_values事先更新对应键的值,否则插入位置依旧是原来的位置

			m_values(成交记录表):
			底层实现是map<IndexID,MarketStreamInfo>,记录所有成交的最新记录信息

			m_stats(成交统计表):

			m_bonds(债券到成交索引的表):
			每个债券可能有多个成交,这多个成交记录的DealIndex构成一个集合
			m_bonds保存每个债券的所有成交,并维护每个债券的最新成交的DealIndex

			3.5.0中新增作废成交,处理逻辑如下:

			*/
		bool stat_erase = nDealStatus == 8 || nDealStatus >= 4;
		if (m_keys->IsExist(id))
		{
			const void* ptr = m_values->FindRecord(id);
			BASSERTS(ptr, "");
			if (ptr)
			{
				const Record& rd = *((const Record*)ptr);
				if (stat_erase)
					m_stats->EraseStat(rd, true);
				else
					m_stats->UpdateStat(rd, info);
			}
			else if (!stat_erase)
			{
				//BLOGERR("[STREAM]invalid DealIndex:%s,does not exist in m_values", id.id.c_str());
				m_stats->AddStat(info);
			}
		}
		else
		{
			if (stat_erase)
				m_stats->EraseStat(info, false);
			else
				m_stats->AddStat(info);
		}

		if (nDealStatus == 8 || nDealStatus < 4)
		{
			m_values->Add(id, &info);
			m_keys->Insert(id);
		}
		else
		{
			m_keys->Erase(id);
			m_values->Remove(id);
		}

		// 如果该成交是作废成交，那么也应该从对应债券的成交id队列中删除，即Remove，而不是之前的Add
		if (nDealStatus < 4)
			m_bonds->Add(info.m_body.m_bondkey, info.m_body.m_listedmarket, info.m_company_id, id, m_values);
		else
			m_bonds->Remove(info.m_body.m_bondkey, info.m_body.m_listedmarket, info.m_company_id, id, m_values);
		m_keys->SetBatchId(batchid);
		m_cache->Add(batchid, &info);
		int companyID = atoi(info.m_company_id);
		std::map<int, int>::iterator it = m_mapversion.find(companyID);
		if (it == m_mapversion.end())
		{
            m_mapversion.insert(std::make_pair(companyID, info.m_indexID));
		}
		else
		{
			if (info.m_indexID > it->second) it->second = info.m_indexID;
		}
		return;

	}

    bool RealTimeDealList::GetStat(const std::string& bondKey, MarketStreamRec& stat)
	{
		return m_stats->GetStat(bondKey, stat);
	}

	bool RealTimeDealList::IsObjectValid()const
	{
		if (m_cache == nullptr || m_keys == nullptr || m_stats == nullptr || m_values == nullptr)
			return false;
		return m_tag == OBJECT_TAG;
	}
	bool RealTimeDealList::ScanForRecord(const char * szBondKey, const  char * szListedMarket, const char* szCompanyId, time_t t, MarketStreamInfo &record)const
	{
		int nMin = 0x7FFFFFFF;
		bool corp_empty = szCompanyId == nullptr || *szCompanyId == '\0';
		//vector<DealIndex> indexes;
		//GetKeys(indexes);
		detail::CBondDealJumpList::KeyIterator it = m_keys->GetBegin();
		//for (int k = 0; k < (int)indexes.size(); k++)
		for (; it != m_keys->GetEnd(); ++it)
		{
			const void* ptr = m_values->FindRecord(*it/*indexes[k]*/);
			if (ptr == nullptr)
				continue;
			const Record& rd = *((const Record*)ptr);
			if (strcmp(rd.m_body.m_bondkey, szBondKey) == 0 && strcmp(rd.m_body.m_listedmarket, szListedMarket) == 0
				&& (corp_empty || strcmp(rd.m_company_id, szCompanyId) == 0))
			{
				if (t <= 0)
				{
					record = rd;
					return true;
				}
				else if (rd.m_body.m_createTime <= t)//大于当前的时间
				{
					int nDelta = t - rd.m_body.m_createTime;
					if (nDelta < nMin)
					{
						nMin = nDelta;
						record = rd;
						return true;
					}
				}
			}
		}
		return false;
	}
	bool RealTimeDealList::FindRecord(const char * szBondKey, const char* szListMarket, const char* szCompanyId, MarketStreamInfo& record, time_t t)const
	{
		// 一个债券可能有多笔成交,rdlist是时间倒序排列,根据时间t来选取一个成交数据
		// ScanForRecord 是
		bool corp_empty = szCompanyId == nullptr || *szCompanyId == '\0';
		if (t <= 0 && !corp_empty && m_bonds)
		{
			DealIndex dindex;
			const void* ptr = nullptr;
			bool suc = false;
			if (m_bonds->Find(szBondKey, szListMarket, szCompanyId, dindex, m_values))
			{
				ptr = m_values->FindRecord(dindex);
				if (ptr)
				{
					record = *((const Record*)ptr);
					suc = true;
				}
			}
			if (m_check)
			{
				MarketStreamInfo msi;
				bool suc_scan = ScanForRecord(szBondKey, szListMarket, szCompanyId, t, msi);
				BASSERTS(suc == suc_scan, "");
				if (suc_scan && suc)
				{
					BASSERTS(memcmp(&msi, &record, sizeof(MarketStreamInfo)) == 0, "");
				}
			}
			return suc;
		}
		return ScanForRecord(szBondKey, szListMarket, szCompanyId, t, record);
	}

	MarketStreamInfo* RealTimeDealList::ElementAt(DealIndex key)
	{
		MarketStreamInfo* pInfo = (MarketStreamInfo*)(m_values->FindRecord(key));
		return pInfo;
	}
	//////////////////////////////////////////////////////////////////////////////////
	//class MultiDayDealList : public RealTimeDealList

	MultiDayDealList::MultiDayDealList(uint32_t dayBegin, uint32_t dayEnd)
		: RealTimeDealList(dayBegin, 0)
	{
		m_dayBegin = dayBegin;
		m_dayEnd = dayEnd;
		if (m_dayBegin > 0 && m_dayEnd > 0 && m_dayBegin <= m_dayEnd)
		{

		}
	}

	MultiDayDealList::~MultiDayDealList()
	{
	}

	uint32_t MultiDayDealList::GetDayBegin()
	{
		return m_dayBegin;
	}
	uint32_t MultiDayDealList::GetDayEnd()
	{
		return m_dayEnd;
	}

	bool MultiDayDealList::CheckDayRange(uint32_t dayBegin, uint32_t dayEnd)
	{
		if (m_dayBegin == dayBegin && m_dayEnd == dayEnd)return true;
		return false;
	}

	void MultiDayDealList::ChangeDayRange(uint32_t beginDay, uint32_t endDay)
	{
		m_dayBegin = beginDay;
		m_dayEnd = endDay;
	}

	void MultiDayDealList::InsertDirectly(const MarketStreamInfo& info, uint32_t unDay, int batchid)
	{
		if (unDay > m_dayEnd || unDay < m_dayBegin)return;

		int nDealStatus = atoi(info.m_body.m_dealStatus);
		DealIndex id(info.m_body.m_id);

		bool stat_erase = nDealStatus == 8 || nDealStatus >= 4;
		if (m_keys->IsExist(id))
		{
			const void* ptr = m_values->FindRecord(id);
			BASSERTS(ptr, "");
			if (ptr)
			{
				const Record& rd = *((const Record*)ptr);
				if (stat_erase)
					m_stats->EraseStat(rd, true);
				else
					m_stats->UpdateStat(rd, info);
			}
			else if (!stat_erase)
			{
				//BLOGERR("[STREAM]invalid DealIndex:%s,does not exist in m_values", id.id.c_str());
				m_stats->AddStat(info);
			}
		}
		else
		{
			if (stat_erase)
				m_stats->EraseStat(info, false);
			else
				m_stats->AddStat(info);
		}

		if (nDealStatus == 8 || nDealStatus < 4)
		{
			m_values->Add(id, &info);
			m_keys->Insert(id);
		}
		else
		{
			m_keys->Erase(id);
			m_values->Remove(id);
		}

		// 如果该成交是作废成交，那么也应该从对应债券的成交id队列中删除，即Remove，而不是之前的Add
		if (nDealStatus < 4)
			m_bonds->Add(info.m_body.m_bondkey, info.m_body.m_listedmarket, info.m_company_id, id, m_values);
		else
			m_bonds->Remove(info.m_body.m_bondkey, info.m_body.m_listedmarket, info.m_company_id, id, m_values);
		m_keys->SetBatchId(batchid);
		m_cache->Add(batchid, &info);
		return;

	}
}
